package org.fastsyncer.manager.framework;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.commons.beanutils.BeanUtils;
import org.fastsyncer.common.entity.Mapping;
import org.fastsyncer.common.entity.MappingTask;
import org.fastsyncer.manager.store.DataCenter;
import org.fastsyncer.worker.framework.WorkerFactory;
import org.json.JSONException;
import org.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class RouterHandle {

	private static Logger logger = LoggerFactory.getLogger(RouterHandle.class);

	private static RouterHandle instance = null;

	private RouterHandle() {
	}

	public static RouterHandle getInstance() {
		if (instance == null) {
			synchronized (RouterHandle.class) {
				instance = new RouterHandle();
			}
		}
		return instance;
	}

	/**
	 * 迁移失败结果
	 */
	private ConcurrentLinkedQueue<Integer[]> syncAll = new ConcurrentLinkedQueue<Integer[]>();

	/**
	 * 同步失败结果
	 */
	private ConcurrentLinkedQueue<Integer> syncInc = new ConcurrentLinkedQueue<Integer>();

	/**
	 * 数据迁移路由
	 * 
	 * @Title: router
	 * @Description: 数据迁移路由
	 * @param mappingTask
	 *            映射关系
	 * @param distributeTask
	 *            拆分的任务
	 * @return: JSONObject 执行结果
	 * @throws Exception
	 */
	public JSONObject router(MappingTask mappingTask, ConcurrentLinkedQueue<Integer[]> distributeTask)
			throws Exception {
		logger.info("Routing...");

		// 目标源驱动
		List<Mapping> mappings = mappingTask.getMappings();
		int dirvers = mappings.size();
		// 线程个数
		CountDownLatch latch = new CountDownLatch(dirvers);
		// 创建线程池
		ExecutorService es = Executors.newCachedThreadPool();
		// 分配配置中设置的线程数
		for (int i = 0; i < dirvers; i++) {
			// 遍历目标源驱动
			MappingTask runTask = new MappingTask();
			BeanUtils.copyProperties(runTask, mappingTask);
			runTask.setTargetMapping(mappings.get(i));

			es.execute(new Runnable() {
				@Override
				public void run() {
					try {
						JSONObject handle = WorkerFactory.getInstance().handle(runTask, distributeTask);
						// 如果有错误
						if (handle.getJSONArray("reason").length() > 0) {
							// 2.1解析错误较耗时，开个线程去存储
							new TaskLogHandle(mappingTask.getId(), handle).start();
						}
						// 获取总数和成功数
						syncAll.offer(new Integer[] { handle.getInt("excute"), handle.getInt("success") });
					} catch (InterruptedException | JSONException e) {
						logger.error("An exception when performing data migration routing tasks:" + e.getMessage(), e.getCause());
					} finally {
						latch.countDown();
					}
				}
			});
		}
		// 使得主线程(main)阻塞直到latch.countDown()为零才继续执行
		latch.await();
		es.shutdown();

		// 统计分发后的执行结果。
		int excute = 0;
		int success = 0;
		// 累加所有目标源的失败数
		while (!syncAll.isEmpty()) {
			Integer[] poll = syncAll.poll();
			excute += poll[0];
			success += poll[1];
		}
		// 总计= 数据源同步总数 * 目标源个数
		excute *= dirvers;

		logger.info("The route ends.");
		JSONObject result = new JSONObject();
		result.put("success", success);
		result.put("fail", excute - success);
		return result;
	}

	/**
	 * 数据同步路由
	 * 
	 * @Title: router
	 * @Description: 数据同步路由
	 * @param mappingTask
	 *            映射关系
	 * @param t
	 *            同步数据
	 * @throws Exception
	 */
	public void router(JSONObject t) throws Exception {
		String taskId = t.getString("taskId");
		MappingTask mappingTask = (MappingTask) DataCenter.getInstance().getMapping(taskId);

		// 目标源驱动
		List<Mapping> mappings = mappingTask.getMappings();
		int dirvers = mappings.size();
		// 线程个数
		CountDownLatch latch = new CountDownLatch(dirvers);
		// 创建线程池
		ExecutorService es = Executors.newCachedThreadPool();
		// 分配配置中设置的线程数
		for (int i = 0; i < dirvers; i++) {
			// 遍历目标源驱动
			MappingTask runTask = new MappingTask();
			BeanUtils.copyProperties(runTask, mappingTask);
			runTask.setTargetMapping(mappings.get(i));

			es.execute(new Runnable() {
				@Override
				public void run() {
					try {
						JSONObject handle = WorkerFactory.getInstance().handle(runTask, t);
						if (handle.getJSONArray("reason").length() > 0) {
							// 记录错误数
							syncInc.offer(handle.getInt("fail"));

							// 开个线程去存储错误日志
							new TaskLogHandle(taskId, handle).start();
						}
					} catch (JSONException e) {
						logger.error("An exception when routing task：" + e.getMessage(), e.getCause());
					} finally {
						latch.countDown();
					}
				}
			});
		}
		// 使得主线程(main)阻塞直到latch.countDown()为零才继续执行
		latch.await();
		es.shutdown();

		// 统计分发后的执行结果。
		int total = t.getJSONArray("msg").length() * dirvers;
		int fail = 0;
		// 累加所有目标源的失败数
		while (!syncInc.isEmpty()) {
			fail += syncInc.poll();
		}

		/**
		 * TODO 这里会遇到针对同一份驱动并发更新操作，导致数据统计不一致问题。
		 * 这里的解决思路：定义一个队列，用于存放同步结果（成功和失败数），然后 去统计队列中成功失败数，将统计好的数据持久化到磁盘。
		 * 这样做法是为了错开频繁更新同一个驱动，提高数据同步并发。但不足的一点是机器发生宕机时，数据有可能丢失，这个是零容忍的!
		 * 所以针对这种情况，有两个解决方案： 1、将同步的结果发送给消息中间件 2、将现在的单节点部署升级为分布式部署
		 * 
		 * PS:目前暂时考虑用队列去实现
		 */
		this.offer(taskId, (total - fail), fail);
	}

	// 存放数据同步成功/失败数
	private final ConcurrentLinkedQueue<Map<String, Object>> QUEUE = new ConcurrentLinkedQueue<Map<String, Object>>();

	/**
	 * 记录数据同步成功/失败数
	 * 
	 * @Title: addTask
	 * @Description: 记录数据同步成功/失败数
	 * @param taskId
	 *            驱动id
	 * @param success
	 *            成功数
	 * @param fail
	 *            失败数
	 * @return: void
	 * @throws JSONException
	 */
	private void offer(String taskId, int success, int fail) throws JSONException {
		Map<String, Object> t = new HashMap<String, Object>();
		t.put("taskId", taskId);
		t.put("success", success);
		t.put("fail", fail);

		if (QUEUE.isEmpty()) {
			// 排号
			QUEUE.offer(t);
			// 触发统计数据同步行结果
			new TouchHandle().start();
		} else {
			// 排号
			QUEUE.offer(t);
		}
	}

	/**
	 * 触发统计数据同步行结果
	 */
	class TouchHandle extends Thread {
	    
	    TouchHandle() {
            setName("fastsyncer touch router thread");
        }
	    
		@Override
		public void run() {
			while (!QUEUE.isEmpty()) {
				try {
					Map<String, Object> poll = QUEUE.poll();
					String taskId = (String) poll.get("taskId");
					int success = (int) poll.get("success");
					int fail = (int) poll.get("fail");

					// 更新驱动执行结果
					MappingTask mappingTask = (MappingTask) DataCenter.getInstance().getMapping(taskId);
					if(mappingTask==null){
						continue;
					}
					mappingTask.setSuccess(mappingTask.getSuccess() + success);
					mappingTask.setFail(mappingTask.getFail() + fail);
					DataCenter.getInstance().storeMapping(taskId, mappingTask);
				} catch (Exception e) {
					logger.error("触发统计数据同步行结果时异常：" + e.getMessage(), e.getCause());
				}
			}
		}

	}

}
